use crate::run_plan::{HandResult, RunPlanPhysicalPlanFuncArg};
use crate::data_frame::DataInstance;
use std::ops::DerefMut;
use crate::physical_plan::PhysicalPlanExpr;
use crate::data_frame::simple_hand_data::SimpleHandData;

pub async fn run_plan_binary_run(
    left: &mut Box<RunPlanPhysicalPlanFuncArg>,
    op: &mut Box<PhysicalPlanExpr>,
    right: &mut Box<RunPlanPhysicalPlanFuncArg>,
    data: &mut DataInstance,
) -> HandResult {
    let left_expr = match left.deref_mut(){
        RunPlanPhysicalPlanFuncArg::Normal(t) => {t.clone()},
        RunPlanPhysicalPlanFuncArg::OtherRunPlan(t) => {
            let mut tmp = Vec::new();
            let r = t.evaluate(data).await;
            for r in r.into_iter(){
                tmp.push(r);
            }
            return tmp;
        },
    };
    let right_expr = match right.deref_mut(){
        RunPlanPhysicalPlanFuncArg::Normal(t) => {t.clone()},
        RunPlanPhysicalPlanFuncArg::OtherRunPlan(t) => {
            let mut tmp = Vec::new();
            let r = t.evaluate(data).await;
            for r in r.into_iter(){
                tmp.push(r);
            }
            return tmp;
        },
    };

    let mut results = Vec::new();
    let args = vec![left_expr, right_expr];

    match data{
        DataInstance::BatchData(ref mut batch_data) => {
            for d in batch_data.iter_mut(){
                let t = op.evaluate(args.as_slice(), d).await;
                results.push(t);
            }
        },
        DataInstance::GroupData(group_data) => {
            for d in group_data.iter_mut(){
                let r = op.evaluate(args.as_slice(), d.data_mut()).await;
                results.push(r);
            }
        },
    }

    return results;
}